package com.example.thread.case4;

import lombok.SneakyThrows;

import java.util.concurrent.*;

public class CompletionServiceDemo {

    static ExecutorService es = Executors.newFixedThreadPool(10);

    /**
     * CompletionService允许应用程序结构化，使用 Producer/Consumer 模式，其中生产者创建任务并提交，消费者请求完成任务的结果并处理这些结果。
     * 将执行服务与类似 Queue 的接口组合，从任务执行中删除任务结果的处理。
     * CompletionService 接口包含用来提交将要执行的任务的 submit() 方法和用来询问下一完成任务的 take()/poll() 方法。
     */
    static CompletionService<String> ecs = new ExecutorCompletionService<String>(es);

    public static void main(String[] args) {

        ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
        int taskCount = 10;
        new Thread(){
            @Override
            public void run() {
                for (int i = 0; i < taskCount ; i++) ecs.submit(new FutureDemo("Task"+i));
            }
        }.start();

        new Thread(){
            @SneakyThrows
            @Override
            public void run() {
                for (int i = 0; i < taskCount ; i++){
                    Future<String> fr = ecs.take();
                    System.out.println(fr.get());
                    resultMap.put(Integer.toString(i), fr.get());
                }
            }
        }.start();

    }

    public static class  FutureDemo implements Callable<String>{

        private final String value;

        public FutureDemo(String var){
            this.value = var;
        }

        @Override
        public String call() throws Exception {
            return Thread.currentThread().getName() + " Hello " + this.value;
        }
    }
}
